package com.renrendoc.network.flow.trigger;

import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public  class CustomTimeWindowTrigger extends Trigger<Object, TimeWindow> {

    private final long interval;

    private final MapStateDescriptor<Long, String> intervalMapState;

    public CustomTimeWindowTrigger(long interval)
    {
        this.interval = interval;
        intervalMapState = new MapStateDescriptor<>("interval_map", Long.class, String.class);
    }

    @Override
    public TriggerResult onElement(
            Object element, long timestamp, TimeWindow window, TriggerContext ctx)
            throws Exception {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            // if the watermark is already past the window fire immediately
            return TriggerResult.FIRE;
        } else {
            MapState<Long, String> partitionedState = ctx.getPartitionedState(intervalMapState);
            long currentTime = System.currentTimeMillis();
            long triggerTime = currentTime - currentTime % (interval * 1000) + interval * 1000 - 1;
            if (null == partitionedState.get(triggerTime)) {
                ctx.registerEventTimeTimer(triggerTime);
                partitionedState.put(triggerTime, "1");
//                System.out.println("set: " + triggerTime + "curr:" + currentTime);
            }

            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                System.out.println(time);
        System.out.println( window.maxTimestamp());
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        // 定时器触发时输出窗口结果
//        System.out.println(time);
//        System.out.println( window.maxTimestamp());
        if (time != window.maxTimestamp()) {
            MapState<Long, String> partitionedState = ctx.getPartitionedState(intervalMapState);
            if (null != partitionedState.get(time)) {
                ctx.deleteEventTimeTimer(time);
                partitionedState.remove(time);
//                System.out.println("remove: " + time);
            }
        }
        return TriggerResult.FIRE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

        ctx.deleteEventTimeTimer(window.maxTimestamp());
    }

}